package com.hps.demo.netty.client;

import com.alibaba.fastjson.JSONArray;
import com.hps.demo.discovery.ConnectManage;
import com.hps.demo.entity.Request;
import com.hps.demo.entity.Response;
import com.hps.demo.netty.codec.JSONDecoder;
import com.hps.demo.netty.codec.JSONEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.net.SocketAddress;
import java.util.Objects;
import java.util.concurrent.SynchronousQueue;

/**
 * @author heps
 * @since 2020/8/15
 */
@Slf4j
@Component
public class NettyClient {

  private EventLoopGroup group = new NioEventLoopGroup(1);

  private Bootstrap bootstrap = new Bootstrap();

  private NettyClientHandler nettyClientHandler;

  private ConnectManage connectManage;

  public NettyClient(NettyClientHandler nettyClientHandler, ConnectManage connectManage) {
    this.nettyClientHandler = nettyClientHandler;
    this.connectManage = connectManage;

    bootstrap.group(group).
            channel(NioSocketChannel.class).
            option(ChannelOption.TCP_NODELAY, true).
            option(ChannelOption.SO_KEEPALIVE,true).
            handler(new ChannelInitializer<SocketChannel>() {
              //创建NIOSocketChannel成功后，在进行初始化时，将它的ChannelHandler设置到ChannelPipeline中，用于处理网络IO事件
              @Override
              protected void initChannel(SocketChannel channel) throws Exception {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast(new IdleStateHandler(0, 0, 30));
                pipeline.addLast(new JSONEncoder());
                pipeline.addLast(new JSONDecoder());
                pipeline.addLast("handler", nettyClientHandler);
              }
            });
  }

  public Object send(Request request) throws InterruptedException {
    Channel channel = connectManage.chooseChannel();
    if (Objects.nonNull(channel) && channel.isActive()) {
      SynchronousQueue<Object> queue = nettyClientHandler.sendRequest(request, channel);
      Object result = queue.take();
      return JSONArray.toJSONString(result);
    } else {
      Response response = new Response();
      response.setCode(1);
      response.setErrorMsg("未正确连接到服务器, 请检查相关配置信息!");
      return JSONArray.toJSONString(response);
    }
  }

  public Channel doConnect(SocketAddress address) throws InterruptedException {
    ChannelFuture future = bootstrap.connect(address);
    return future.sync().channel();
  }
}
